package mq;

import org.springframework.util.StringUtils;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;


class Resource{
    public volatile boolean flag = true;

    private AtomicInteger atomicInteger = new AtomicInteger();

    private BlockingQueue<String> blockingQueue;

    public Resource(BlockingQueue<String> blockingQueue){
        this.blockingQueue = blockingQueue;
        System.out.println(blockingQueue.getClass().getName());
    }

    public void product() throws InterruptedException {
        String data;
        while(flag){
            data = atomicInteger.incrementAndGet()+"";
            boolean offer = blockingQueue.offer(data, 2, TimeUnit.SECONDS);
            if(offer){
                System.out.println(Thread.currentThread().getName()+"\t插入队列数据"+data+"成功");
            }else{
                System.out.println(Thread.currentThread().getName()+"\t插入队列数据"+data+"失败");
            }
            TimeUnit.MILLISECONDS.sleep(100);
        }
        System.out.println(Thread.currentThread().getName() + "\t 停止 表示 flag=" + flag);
    }

    public void consume() throws InterruptedException {
        String result = null;
        while (flag){
            result = blockingQueue.poll(2L, TimeUnit.SECONDS);
            if(StringUtils.isEmpty(result)){
                flag = false;
                System.out.println(Thread.currentThread().getName()+"超时没有取到数据 关闭消费者");
                System.out.println();
                System.out.println();
                return;
            }
            System.out.println(Thread.currentThread().getName() + "消费队列" + result + "成功");
        }
    }

    public void stop() throws Exception {
        flag = false;
    }
}


public class ProdConsumerBlockQueueDemo {
    public static void main(String[] args) throws Exception {
        Resource resource = new Resource(new ArrayBlockingQueue<>(10));
        new Thread(() -> {
            try {
                resource.product();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"Product").start();

        new Thread(() -> {
            try {
                resource.consume();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"Consume").start();

        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println();
        System.out.println();
        System.out.println();
        System.out.println("时间到,停止活动");
        resource.stop();

    }
}